<?php namespace xlx;

use Workerman\Connection\TcpConnection;
use \Workerman\Events\EventInterface;

class DanbooActor extends \xlx\Actor 
{  
    public static $addr = 'tcp://127.0.0.1:55854';
    public static $protocol =  '\xlx\Protocols\Frame'; 
    public static $selfname='';

    public static $server;
    public static function startClient($event=null){    
        $socket = \stream_socket_client(DanbooActor::$addr) or exit('??'); 
        self::$server = new TcpConnection($socket);
        self::$server->protocol = DanbooActor::$protocol;    
        self::$server->onMessage = function($conn,$data){    
            switch ($data[0]) {
                case 'send':     
                    if(self::exists($to)){
                        parent::sendTo($data[2],$data[3]);
                        self::sendData($data[2],$data[1],['callback'=>'send','return'=>'ok']); 
                    }else{
                        self::sendData($data[2],$data[1],['callback'=>'send','throw'=>'exit']);  
                    }
                    //parent::mailto($data[1],$data[2]);
                    // $actor = self::$act ors[$data[1]];
                    // $actor->mailbox[]=$data[2];
                    // $actor->not ify(); 
                    break; 
                case 'monitor':
                    parent::sendTo($data[1],$data[2]);
                    //self::watch($data[1],'');
                    break;
                case 'exit':
                    //self::throw($data[1]); 
                    //self::notify($data[1]);
                    // $actor = self::$actors[$data[1]];
                    // $actor->exit = true;
                    // $actor->notify();
                default: 
                    break;
            }  
        };
        self::$server->onClose=function($conn){ 
            self::$server=null;
        };   
        self::$selfname = (string)microtime(true);// \stream_socket_get_name( self::$server->getSocket(), false);
        self::$server->send(['onStart',self::$selfname]); 
    } 
    public function __construct(&$pid=null){   
        $pid = self::$selfname.':'.(string)microtime(true) ; 
        // return $this->receive(function($data){
        //     if( isset($data['throw']) && $data['throw']=='exit'){ 
        //         exit;
        //     } 
        // });
        parent::__construct($pid);
    }
    public function __destruct(){ 
        if(isset(self::$server)){
            self::$server->send(['onExit',$this->self() ]);
        }else{
            $this->onExit(null, $this->self() );
        }  
        parent::__destruct();
    }
    protected static function sendData($from,$to,$data){    
        if(self::$server){
            self::$server->send(['onSend',$from,$to,$data]);   
        }else{
            self::onSend(null,$from,$to,$data);
        }  
    }  

    public function send($to,$data){   
        if(isset(self::$regs[$to])){
            $to = self::$regs[$to];
        } 
        if(isset(self::$server)){    
            self::$server->send(['onSend',$this->self(),$to,$data ]);  
        }else{
            self::onSend(null,$this->self(),$to,$atom); 
        }   
        return $this->receive(function($data){
            if(isset($data['return']) && isset($data['callback']) && $data['callback']=='send'){ 
                return $data['return']; 
            }
            if(isset($data['throw']) && isset($data['callback']) && $data['callback']=='send'){ 
                throw new \Exception($data['throw']); 
            }
        });
    }
    public function register($atom){ 
        if(isset(self::$server)){ 
            self::$server->send(['onRegister',$this->self(),$atom ]);  
        }else{
            self::onRegister(null,$this->self(),$atom); 
        } 
        return $this->receive(function($data){
            if(isset($data['return']) && isset($data['callback']) && $data['callback']=='register'){ 
                return $data['return']; 
            }
            if(isset($data['throw']) && isset($data['callback']) && $data['callback']=='send'){ 
                throw new \Exception($data['throw']); 
            }
        });
    }
    public function unregister($atom){        
        if(isset(self::$server)){ 
            self::$server->send(['onUnRegister',$atom ]);  
        }else{
            self::onUnRegister(null,$atom );
        }  
    }


    public function monitor($to){
        if($to instanceof Actor){
            $to = $to->pid; 
        }     
        if(isset(self::$server)){ 
            self::$server->send(['onMonitor',$to, $this->self() ]);  
        }else{
            self::onMonitor(null,$to,$this->self() );
        }   
        return $this->receive(function($data){
            if(isset($data['return']) && isset($data['callback']) && $data['callback']=='register'){ 
                return $data['return']; 
            }
        });
    }
    public function demonitor($to){

    }

    public function link($to){

    }
    public function unlink($to){

    } 
 

    /************  server  *************/
    public static $watch;
    public static $conns; 
    public static $regs; 
    public static function startServer($event){ 
        $socket = \stream_socket_server(DanbooActor::$addr, $errno, $errstr); 
        $event->add($socket, EventInterface::EV_READ, function ($socket)  {
            $socket = \stream_socket_accept($socket, 0, $addr);
            $client = new TcpConnection($socket,$addr); 
            $client->protocol = DanbooActor::$protocol; 
            $client->onMessage = function($conn,$data){    
                $fn = $data[0]; $data[0] = $conn;
                \call_user_func_array([static::class,$fn],$data); 
            };
            $client->onClose=function($conn) { 
                unset(self::$conns[$conn->name]);
            };
        });
        self::$selfname =  '0000000000.0000';//\stream_socket_get_name( self::$server->getSocket(), false);  
    } 

    public static function onStart($conn,$name){
        $conn->name = $name;
        self::$conns[$name]=$conn;  
    }
    public static function onRegister($conn,$from,$atom){
        if(isset(self::$regs[$atom])){ 
            self::sendData($from,$from,['callback'=>'register','return'=>false]); 
        }else{
            self::$regs[$atom]=$from;
            self::sendData($from,$from,['callback'=>'register','return'=>true]); 
        }
    }
    public static function onUnRegister($conn,$atom){
        unset(self::$regs[$atom]);
    }
    public static function onSend($conn,$from,$to,$data){
        if(isset(self::$regs[$to])){
            $to = self::$regs[$to];
        } 
        if(strstr($to,':')){
            list($local,$id) = explode(':',$to,2);
            if(isset(self::$conns[$local])){
                self::$conns[$local]->send(['send',$from,$to,$data]); 
                return;
            }else{ 
                if($conn){
                    if(parent::sendTo($to,$data)){
                        self::sendData($to,$from,['callback'=>'send','return'=>'ok']); 
                    }else{
                        self::sendData($to,$from,['callback'=>'send','throw'=>'exit']);  
                    }
                }else{
                    parent::sendTo($to,$data);   
                }
                return;
            }  
            //if error
            print_r([$to,$data,array_keys(self::$conns)]);
        }
    }
    public static function onMonitor($conn,$to,$pid){
        if(isset(self::$regs[$to])){
            $to = self::$regs[$to];
        } 
        if(strstr($to,':')){
            list($local,$id) = explode(':',$to,2);
            if(isset(self::$conns[$local])){ 
                self::$conns[$local]->send(['monitor',$to,$pid]); 
                self::$watch[$to][] = $pid;
                return;
            }
        }
        throw new Exception("Error Processing Request", 1);
        
    }
    public static function onDemonitor($conn,$to,$pid){
        $index = array_search($pid,self::$watch[$to]);
        if($index>=0){
            array_splice(self::$watch[$to],$index,1);
        }
    } 
    public static function onExit($conn,$pid){
        // echo ":::exit:::\n";
        // print_r(self::$watch);
        if(empty(self::$watch[$pid]))
            return ;
        foreach (self::$watch[$pid] as $p) { 
            list($local,$id) = explode(':',$p,2);
            if(isset(self::$conns[$local])){
                self::$conns[$local]->send(['exit',$p]); 
                continue;
            }else{ 
                self::throw($p);
                //self::notify($p);
             //   parent::mailto($p,['exit'] );
            // if(isset(self::$actors[$to])){
            //     $actor = self::$actors[$to];
            //     $actor->exit = true; 
            //     $actor->notify();
            //     continue;
            }  
            //if error
            print_r([$to,$data,array_keys(self::$conns)]);
        }
    }

}



// class DanboActorClient extends DanbooActor
// {  
//     public function send($args){ 
//         $this->conn->send([$args]);   
//         return $this;
//     } 
//     public function receive($receive){ 
//         $defer = $this->defers[] = new \React\Promise\Deferred();
//         $defer->receive = $receive;
//         return $defer->promise();
//     }  
// }

// class DanbooActorServer extends DanbooActor
// { 

//     public static function createServer($event){ 
//         $server = \stream_socket_server(DanbooActor::$addr, $errno, $errstr);
//         $event->add($server, EventInterface::EV_READ, function ($socket)use($class,$protocol) {
//             $socket = \stream_socket_accept($socket, 0, $addr);
//             $client = new TcpConnection($socket,$addr); 
//             $client->protocol = $protocol; 
//             $client->onMessage= function($conn,$data){     
//                 $promise = \xlx\co([self::class,"call_{$data[1]}"],...$data[2]);    
//                 $promise->then(function($result)use($data, $conn ){   
//                     $conn->send( [$data[0], $result ]);  
//                 },function($e)use($data, $conn){   
//                     $conn->send( [$data[0], null , $e->getMessage()]);  
//                 });    
//             };
//             $client->onClose=function($conn)use($class,$addr){
//                 $class::onClose($conn,$addr);
//             };
//             $class::onConnect($client,$addr);
//         });
//     }
//     public function send($name,$args){
//         self::$server->send([\microtime(), $name, $args]);   
//     } 
//     public function receive($receive){ 
//         $defer = $this->defers[] = new \React\Promise\Deferred();
//         $defer->receive = $receive;
//         return $defer->promise();
//     } 

// }